package com.ruoyi.dlxx.kafka;


import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerApplication {

    private static final String TOPIC="topic_test_xh"; //"BA4035_BA4035010001_mhhwclxsgjb";//
    private static final String CONFIG_DIR="D:\\JavaProject\\kafka\\fanmi\\";
    private static final String BOOTSTRAP_SERVERS="11.100.5.102:32610,11.100.5.101:32283,11.100.5.99:31994";
    private static final String TOS_PRINCIPAL="tos_k2142rz";

    public static void main(String[] args) {
        SpringApplication.run(KafkaConsumerApplication.class, args);

        try {
            // 集群krb5.conf文件配置
            System.setProperty("java.security.krb5.conf", CONFIG_DIR + "krb5.conf");
            // 集群jaas.conf文件配置
            System.setProperty("java.security.auth.login.config", CONFIG_DIR + "jaas.conf");
//       System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
            // System.setProperty("sun.security.krb5.debug", "true");
//
//       // 集群kafka bootstrap-server
            Properties props = new Properties();
            props.put("bootstrap.servers", BOOTSTRAP_SERVERS);

            props.put("sasl.kerberos.service.principal.instance", TOS_PRINCIPAL);
//           //指定kafka消费者组 命名规则 账号名
            props.put("group.id", "default");
            props.put("enable.auto.commit", "false");
            props.put("auto.commit.interval.ms", "1000");// acks
//            props.put("auto.offset.reset", "earliest");// retries
            props.put("auto.offset.reset", "earliest");// retries
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

            props.put("security.protocol", "SASL_PLAINTEXT");// security.protocol
            props.put("sasl.mechanism", "GSSAPI");// sasl.mechanism
            props.put("sasl.kerberos.service.name", "kafka");// sasl.kerberos.service.name
            System.out.println("读取配置信息成功！");

//            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
//            System.out.println("开始创建");
//            consumer.subscribe(Arrays.asList(TOPIC));
//            System.err.println("----------开始消费------------");
//            while (true) {
//                ConsumerRecords<String, String> records = consumer.poll(10);
//                for (ConsumerRecord<String, String> record : records) {
//                    System.out.println("offset =" + record.offset() + ",value =" + record.value());
//                    System.err.println();
//                }
//            }
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            System.out.println("读取配置信息失败！");
        }
    }

}
